package com.boarsoft.message.batch;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.boarsoft.common.util.DateUtil;
import com.boarsoft.message.bean.Message;
import com.boarsoft.message.bean.TargetMQ;
import com.boarsoft.message.core.MessageContext;
import com.boarsoft.message.store.jdbc.MsgJdbcRoller;

/**
 * 
 * @author Mac_J
 *
 */
public class MsgJdbcTableCut implements Runnable {
	private static final Logger log = LoggerFactory.getLogger(MsgJdbcTableCut.class);

	@Autowired
	private MessageContext messageContext;

	protected ScheduledExecutorService scheduler;

	/** 配置信息map */
	protected Map<String, MsgTableCutCfg> configMap;

	/** 默认30分钟检查一次 */
	protected long interval = 30l;

	private SimpleDateFormat f = new SimpleDateFormat("yyyyMMddHHmm");

	@PostConstruct
	public void init() {
		if (scheduler == null) {
			return;
		}
		scheduler.scheduleWithFixedDelay(this, interval, interval, TimeUnit.MINUTES);
	}

	@Override
	public void run() {
		Map<String, TargetMQ> tqMap = messageContext.getTargetQueueMap();
		log.info("Cut table of target message queue size:{}", tqMap.size());
		for (String target : tqMap.keySet()) {
			TargetMQ tq = tqMap.get(target);
			MsgJdbcRoller roller = (MsgJdbcRoller) tq.getRoller();
			log.info("Cut table of target message roller: {}", roller);
			if (roller == null) {
				continue;
			}
			MsgTableCutCfg cutCfg = configMap.get(target);
			if (cutCfg == null) {
				cutCfg = new MsgTableCutCfg();
				configMap.put(target, cutCfg);
			}
			try {
				this.cut(target, roller, cutCfg);
			} catch (Exception e) {
				log.error("Error on cut table of target message queue {}", target, e);
			}
		}
	}

	/**
	 * 将工作表中N小时前的数据移到新建的备份表<br>
	 * 并删除最老的一张备份表（始终只保留M张备份表）<br>
	 * 此批量在工作表中过期数据量超过单表上限时触发<br>
	 * 注：工作表以消息ID为主键，登记地址、创建日期、状态 为索引<br>
	 * 工作表和备份表都使用MyISAM引擎(目前用的是InnoDB) 只有main表才备份，detail表只管清除过期数据
	 * 备份表只有达到最大容量时才会新建备份表，表名结尾加上当前时间戳（精确到小时）
	 * 备份main表时，每次备份5000条，然后删除已备份的数据，不足5000则不备份，等到下次检查达到5000才继续备份
	 * 
	 * @throws Exception
	 */
	public void cut(String target, MsgJdbcRoller roller, MsgTableCutCfg cutCfg) throws Exception {
		log.warn("Cut table of target message queue {}", target);
		// 获取当前时间（精确到小时）
		String currentTimeStr = f.format(new Date());
		Date currentTime = f.parse(currentTimeStr);
		// 计算N小时以前的时间（精确到小时）
		Date beforeTime = DateUtil.addMinute(currentTime, -(cutCfg.getExpiredTime()));

		// 不需要备份则直接删除过期数据后返回
		if (!cutCfg.isNeedBackup()) {
			// 删除工作表中的N小时以前的过期数据
			roller.deleteExpiredData(beforeTime.getTime(), Message.STATUS_SENT);
			return;
		}

		// 以此时间为条件，统计工作表中过期数据量
		int expiredNum = roller.expiredNum(beforeTime, Message.STATUS_SENT, Message.STATUS_DONE, Message.STATUS_FAILED,
				Message.STATUS_INVALID);
		log.info("expired message num :{},beforeTime:{}", expiredNum, beforeTime.getTime());
		// 没有过期数据
		if (expiredNum == 0) {
			log.info("No expired message in table of {}", target);
			return;
		}

		// 以当前时间为条件，（分批次？存储过程？）以追加模式复制工作表中N小时以前的的数据到备份表
		int pageNum = expiredNum / cutCfg.getBatchSize();
		int result = 0;
		// 过期数据量小于dataNum则不备份
		while (expiredNum / cutCfg.getBatchSize() >= 1) {
			// 检查备份表容量
			String backup = checkBackupTab(roller, cutCfg);
			result = roller.moveExpiredData(backup, cutCfg.getBatchSize(), beforeTime.getTime(), Message.STATUS_SENT,
					Message.STATUS_DONE, Message.STATUS_FAILED, Message.STATUS_INVALID);
			log.info("{} move data num :{}", pageNum--, result);
			// 如果迁移的数据量等于dataNum，说明成功
			if (result == cutCfg.getBatchSize()) {
				// 删除工作表中本批次已备份数据
				roller.deleteExpiredData(beforeTime.getTime(), backup, Message.STATUS_SENT, Message.STATUS_DONE,
						Message.STATUS_FAILED, Message.STATUS_INVALID);
			} else {
				// 如果复制失败，则根据时间条件，删除这批次已经导入的数据
				roller.deleteBackupData(beforeTime.getTime(), backup);
			}
			expiredNum -= cutCfg.getBatchSize();
		}
		// 检查备份表的数量，如果超过M张，则删除掉最老的一张备份表
		int backupTabNum = roller.countBackupTabNum();
		if (backupTabNum > cutCfg.getBackupTables()) {
			// 删除最老的备份表
			roller.deleteOldestBackupTab();
		}
	}

	private String checkBackupTab(MsgJdbcRoller roller, MsgTableCutCfg cutCfg) throws Exception {
		// 检查备份表是否存在，没有则创建
		String backup = roller.backupExists();
		if (backup != null) {
			// 如果备份表已存在，则统计备份表的数据量
			int backupDataNum = roller.backupDataNum(backup);
			// 计算导入后的总数据量
			int sum = cutCfg.getBatchSize() + backupDataNum;
			log.info("expired message num add backup data num:{}", sum);

			// 如果导入后的总数据量超过单表最大容量，则备份表为归档表，重建备份表
			if (sum > cutCfg.getTableCapacity()) {
				String currentTimeStr = f.format(new Date());
				log.info("create new table time :{}", currentTimeStr);
				roller.backupCreate(currentTimeStr);
				backup = roller.getBackupTabName(currentTimeStr);
				log.info("create backup table {}", backup);
			}
		} else {
			String currentTimeStr = f.format(new Date());
			roller.backupCreate(currentTimeStr);
			backup = roller.getBackupTabName(currentTimeStr);
			log.info("create backup table {}", backup);
		}
		return backup;
	}

	public Map<String, MsgTableCutCfg> getConfigMap() {
		return configMap;
	}

	public void setConfigMap(Map<String, MsgTableCutCfg> configMap) {
		this.configMap = configMap;
	}

	public long getInterval() {
		return interval;
	}

	public void setInterval(long interval) {
		this.interval = interval;
	}

	public ScheduledExecutorService getScheduler() {
		return scheduler;
	}

	public void setScheduler(ScheduledExecutorService scheduler) {
		this.scheduler = scheduler;
	}

}
